package com.junjie.food.restaurantServiceManager.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.junjie.food.restaurantServiceManager.dao.ProductDao;
import com.junjie.food.restaurantServiceManager.dao.RestaurantDao;
import com.junjie.food.restaurantServiceManager.dto.OrderMessageDTO;
import com.junjie.food.restaurantServiceManager.enummeration.ProductStatus;
import com.junjie.food.restaurantServiceManager.enummeration.RestaurantStatus;
import com.junjie.food.restaurantServiceManager.po.ProductPO;
import com.junjie.food.restaurantServiceManager.po.RestaurantPO;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

@Slf4j
@Service
public class OrderMessageService {

    private ObjectMapper objectMapper = new ObjectMapper();

    @Autowired
    private ProductDao productDao;
    @Autowired
    private RestaurantDao restaurantDao;

    @Autowired
    private Channel channel;

    @Async
    public void handleMessage() throws IOException, TimeoutException, InterruptedException {
        log.info("start linstening message");

        // 定义死信队列
        channel.exchangeDeclare("exchange.dlx"
                ,BuiltinExchangeType.TOPIC
                ,true
                ,false
                ,null);

        // 定义死信队列
        channel.queueDeclare("queue.dlx",
                true,
                false,
                false,
                null);

        // 队列和交换机绑定
        // 这里使用 # 就表示所有消息都可以入到该队列中
        channel.queueBind("queue.dlx","exchange.dlx","#");

        channel.exchangeDeclare(
                "exchange.order.restaurant",
                BuiltinExchangeType.DIRECT,
                true,
                false,
                null);

        // 尽量指定初始大小，因为扩容是非常耗费资源的
        HashMap<String, Object> args = new HashMap<>(16);
        // 设置整个队列中的消息过期时间都是 150秒
//        args.put("x-message-ttl",150000);
        // 这个参数千万不要设置，这个是队列过期时间，如果队列一段时间没消息就把队列删了
//        args.put("x-expire",15000);

        // 声明死信队列的参数 第二个参数是指定的自定义死信交换机
//        args.put("x-dead-letter-exchange","exchange.dlx");
        // 设置队列最长长度，如果还设置了死信，那么多出来的消息就会放入死信队列中
//        args.put("x-max-length",5);

        // 将参数塞入队列中
        // 这里注意，如果启动后报错，说该值为none
        // 那就去rabbitMQ控制台手动删除该队列，再重启该项目重新创建队列
        // 因此如果是要修改队列属性的话，切记一定要先删除该队列，再重新创建一个
        channel.queueDeclare(
                    "queue.restaurant",
                    true,
                    false,
                    false,
                args);

            channel.queueBind(
                    "queue.restaurant",
                    "exchange.order.restaurant",
                    "key.restaurant");

            // 最多同时可以有两条消息是在处理中，没有被ack的状态，如果超过了就限流暂时不发消息过来
            // 而且这时候就可以直接再启动一个商家微服务去抢消息
//            channel.basicQos(2);

            // 这里第二个参数就是设置是否为 自动ACK  因为我们要使用手动的，所以设置成false
            channel.basicConsume("queue.restaurant", false, deliverCallback, consumerTag -> {
            });
            while (true) {
                Thread.sleep(100000);
            }

    }

    // 餐厅消费者
    DeliverCallback deliverCallback = (consumerTag, message) -> {
        String messageBody = new String(message.getBody());
        log.info("deliverCallback:messageBody:{}", messageBody);

        try {
            OrderMessageDTO orderMessageDTO = objectMapper.readValue(messageBody,
                    OrderMessageDTO.class);
            // 查询该产品
            ProductPO productPO = productDao.selsctProduct(orderMessageDTO.getProductId());
            log.info("onMessage:productPO:{}", productPO);
            // 查询该商铺
            RestaurantPO restaurantPO = restaurantDao.selsctRestaurant(productPO.getRestaurantId());
            log.info("onMessage:restaurantPO:{}", restaurantPO);
            // 判断产品是否可用，商户是否营业 来修改消息类里面的一些内容
            if (ProductStatus.AVALIABLE == productPO.getStatus() && RestaurantStatus.OPEN == restaurantPO.getStatus()) {
                // 确认订单并填写价格
                orderMessageDTO.setConfirmed(true);
                orderMessageDTO.setPrice(productPO.getPrice());
            } else {
                orderMessageDTO.setConfirmed(false);
            }
            log.info("sendMessage:restaurantOrderMessageDTO:{}", orderMessageDTO);


                // 当消息无法被路由的时候，就会回调该方法
//                channel.addReturnListener(new ReturnListener() {
//                    @Override // 参数全放在形参上
//                    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
//                        // 除了打印log，还可以加别的业务操作，这里就懒得全部打印了
//                        log.info("Message Return,replyCode:{}:",replyCode);
//                    }
//                });

                // ReturnCallBack，只是参数封装在对象里面，看起来舒服点
                channel.addReturnListener(new ReturnCallback() {
                    @Override // 这个和前面接口的区别就是，参数全封装在对象里面
                    public void handle(Return aReturn) {
                        log.info("Message Return");
                    }
                });

                // 稍微休眠一下，模拟处理业务
//                Thread.sleep(3000);
                // 手动签收消息 这里细节是，我们必须使用收消息的channel进行签收！！
                // 因此这里我们将channel变成了成员变量，这样就线程共享了
                // 第一个参数是发送序号  第二个参数为false表示单条
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);

                // 模仿死信队列的情况，就是手动拒收消息且不让回到该队列 那么该消息就变成了死信
//                channel.basicNack(message.getEnvelope().getDeliveryTag(),false,false);

                // 一次性手动签收5条消息
//                if(message.getEnvelope().getDeliveryTag()%5==0)
//                channel.basicAck(message.getEnvelope().getDeliveryTag(),true);

                // 手动拒收消息 第三个参数设置为true，但是让该消息重回队列
//                channel.basicNack(message.getEnvelope().getDeliveryTag(),false,true);

            AMQP.BasicProperties messageProperties = message.getProperties();
            String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);

            // 将消息又发给订单微服务
            // 这里设置了一个true，意思是开启消息返回确认机制，如果没有成功被路由，会调用ReturnListener
            channel.basicPublish("exchange.order.restaurant",
                    "key.order", true,messageProperties, messageToSend.getBytes());

            // 线程休眠，避免channel走出try直接被close，然后收不到回调消息
            Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

    };
}

